#include "StdAfx.h"
#include "JsonStreamServer.h"
#include "json.h"
#include "MemoryManager.h"
#include "Types.h"
#include "Element.h"
#include "TimestampGenerator.h"
#include "DocumentIdentifierGenerator.h"
#include "QueueEntity.h"
#include "RelationSynopsis.h"
#include "LineageSynopsis.h"
#include "JsonStreamServer.h"
#include "JsonSchema.h"
#include "SchemaInterpreter.h"
#include "Operator.h"
#include "RowWindowOperator.h"
#include "JoinOperator.h"
#include "SelectionOperator.h"
#include "ProjectionOperator.h"
#include "RootOperator.h"
#include "RstreamOperator.h"
#include "IstreamOperator.h"
#include "DstreamOperator.h"
#include "LeafOperator.h"
#include "Scheduler.h"
#include "IOManager.h"
#include "SocketStreamInput.h"
#include "IStreamInput.h"
#include "RandomGeneratedStreamInput.h"
#include "ShowResultStreamOutput.h"
#include "IStreamOutput.h"
#include "QueryManager.h"
#include "QueryIntermediateRepresentation.h"
#include "JsonSchema.h"
#include "PlanManager.h"
#include "SchemaManager.h"
#include "CommandManager.h"
#include <windows.h>
JsonStreamServer* JsonStreamServer::jsonStreamServer = NULL;
JsonStreamServer::JsonStreamServer(void)
{
	
}


JsonStreamServer::~JsonStreamServer(void)
{

}
JsonStreamServer* JsonStreamServer:: getInstance(void)
{
	if(jsonStreamServer == NULL)
	{
		jsonStreamServer = new JsonStreamServer();
	}
	return jsonStreamServer;
}
void JsonStreamServer::initial()
{
	IOManager::getInstance();
	MemoryManager::getInstance();
	PlanManager::getInstance();
	QueryManager::getInstance();
	SchemaManager::getInstance();
	CommandManager::getInstance();
}
//void JsonStreamServer::endSpecification()
//{
// 
//	initialScheduler();
//
//}
//void JsonStreamServer::initialScheduler()
//{
//	this->scheduler.reset(new Scheduler(PlanManager::getInstance()->getOperatorList()));
// 
//}


void JsonStreamServer::registerStream(boost::shared_ptr<IStreamInput> streamInput)
{
	 QueryManager::getInstance()->addStreamInput(streamInput);
}
void JsonStreamServer::registerQuery(boost::shared_ptr<QueryIntermediateRepresentation> queryIntermediateRepresentation,boost::shared_ptr<IStreamOutput> streamOutput)
{
	QueryManager::getInstance()->addStreamOutput(streamOutput);
	PlanManager::getInstance()->generateQueryPlan(queryIntermediateRepresentation);
	Scheduler::getInstance()->setOperatorList(PlanManager::getInstance()->getOperatorList());
}
void JsonStreamServer::execute()
{
	//for(int i = 0 ; i<27 ;i++)
	while(1)
	{
		IOManager::getInstance()->executeOnce();
		boost::shared_ptr<Operator> op;
		int executionTimes;
		if(!Scheduler::getInstance()->isEmpty())
		{
			Scheduler::getInstance()->getNextOperatorToExecute(op,executionTimes);
			op->execution(executionTimes);

		}
		//Sleep(2000);
	}
}